package com.ht.api.db.hlink;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.springframework.stereotype.Component;

import com.ht.api.db.Db;
import com.ht.api.db.DbConfig;
import com.ht.api.db.Entity;
import com.ht.api.db.hlink.filter.FilterExpression;
import com.ht.api.db.hlink.filter.FilterGroup;
import com.ht.api.db.hlink.filter.JsonConditionEvaluator;
import com.ht.api.db.hlink.function.HlinkFunction;
import com.ht.api.db.task.HtTaskExecutor;
import com.ht.api.db.util.FunctionUtil;
import com.ht.api.db.util.HtUtil;
import com.ht.api.db.util.RedisUtil;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * 源数据过滤器
 * @author asq
 * @createTime 2024年12月13日 09:52:01
 */
@Slf4j
@Component
public class Hlink {
	/**
	 * 根据表名执行一批数据的映射
	 * @param ds 数据源
	 * @param tableCode 表名
	 * @param dataList 数据列表
	 */
	public static void execute(DataSource ds, String tableCode, List<Entity> dataList) {
		// open_data表和普通表，走不同的逻辑
		if ("open_data".equals(tableCode)) {
			// open_data表需要根据source_code进行分组
			dataList.stream().collect(Collectors.groupingBy(openData -> openData.getStr("source_code")))
				.entrySet().stream().forEach(groupSubOpenData -> {
					List<Entity> subList = groupSubOpenData.getValue();
					String sourceCode = subList.get(0).getStr("source_code");
					executeByOpenData(ds, tableCode, sourceCode, subList);
				});
		} else {
			executeByTable(ds, tableCode, dataList);
		}
	}
	
	/**
	 * 根据普通表名（非open_data表）执行一批数据的映射
	 * @param ds 数据源
	 * @param tableCode 表名
	 * @param dataList 数据列表
	 */
	private static void executeByTable(DataSource ds, String tableCode, List<Entity> dataList) {
		// 获取映射列表
		List<Entity> relationList = RedisUtil.get("hlink_relation" + tableCode);
		if (relationList == null || relationList.isEmpty()) {
			return;
		}
		
		// 执行hlink映射		
		executeByRelationList(ds, tableCode, relationList, dataList);
	}
	
	/**
	 * 根据open_data表中的source_code执行一批数据的映射
	 * @param ds 数据源
	 * @param tableCode 源表名
	 * @param sourceCode
	 * @param dataList
	 */
	private static void executeByOpenData(DataSource ds, String tableCode, String sourceCode, List<Entity> dataList) {
		// 获取映射列表
		Entity hlinkSource = RedisUtil.get("hlink_source" + sourceCode);
		if (hlinkSource == null) {
			log.info("[Hlink]没有hlink_source，跳过sourceCode：{}", sourceCode);
			return;
		}
		List<Entity> relationList = RedisUtil.get("hlink_relation" + hlinkSource.getMainId());
		if (relationList == null || relationList.isEmpty()) {
			log.info("[Hlink]没有hlink_relation，跳过sourceCode：{}", sourceCode);
			return;
		}
		
		// 从open_data的data字段中解析源数据列表
		List<Entity> sourceDataList = new ArrayList<>();
		for (Entity openData : dataList) {
			sourceDataList.add(openData.getJSONObject("data").toBean(Entity.class));
		}
		
		// 执行hlink映射
		executeByRelationList(ds, tableCode, relationList, sourceDataList);
	}
	
	/**
	 * 根据映射规则列表对一批源数据列表执行hlink映射
	 * @param ds 数据源
	 * @param tableCode 源表名
	 * @param relationList 映射规则列表
	 * @param dataList 源数据列表
	 */
	private static void executeByRelationList(DataSource ds, String tableCode, List<Entity> relationList, List<Entity> dataList) {

		relationList.stream().forEach(relation -> {
			// 根据配置，使用同步或者异步的方式
			Boolean isSync = relation.getBool("sync");
			if (isSync != null && isSync) {
				doLinkThenSave(ds, tableCode, relation, dataList);
			} else {
				HtTaskExecutor.execute(() -> {
					doLinkThenSave(ds, tableCode, relation, dataList);
				});
			}
		});
	}

	/**
	 * 根据映射规则获取映射后的目标数据列表，然后保存数据库
	 * @param ds
	 * @param tableCode 源表名
	 * @param relation 映射规则
	 * @param dataList 映射前的源数据列表
	 */
	private static void doLinkThenSave(DataSource ds, String tableCode, Entity relation, List<Entity> dataList) {
		List<Entity> dbList = getHlinkDataList(ds, tableCode, relation, dataList);
		Db.insertBatchWithStream(DbConfig.getStreamLoadConfigByDataSource(ds), dbList);
	}

	/**
	 * 获取hlink映射后的目标数据列表
	 * @param ds 数据源
	 * @param tableCode 源表名
	 * @param relation 映射规则列表
	 * @param dataList 源数据列表
	 * @return
	 */
	private static List<Entity> getHlinkDataList(DataSource ds, String tableCode, Entity relation, List<Entity> dataList) {
		JSONObject config = JSONUtil.parseObj(relation.getStr("config"));
		String targetTableCode = relation.getStr("target_table_code");
		List<String> dateTimeColumns = DbConfig.getDateTimeColumns(targetTableCode);
		List<Entity> dbList = new ArrayList<>();
		dataList.stream().forEach(o -> {
			dbList.addAll(getHlinkEntityListByMapper(ds, dateTimeColumns, config, o, targetTableCode));
		});
		log.info("[Hlink]执行映射，{}({}条) >> {}({}条)", tableCode, dataList.size(), targetTableCode, dbList.size());
		return dbList;
	}
	
	/**
	 * 根据mapper映射规则，获取映射后的数据列表
	 * @param ds 数据源
	 * @param dateTimeColumns 需要格式化的时间字段
	 * @param config 映射规则
	 * @param data 源数据
	 * @param tableCode 表名
	 */
	private static List<Entity> getHlinkEntityListByMapper(DataSource ds, List<String> dateTimeColumns, 
			JSONObject config, JSONObject data, String tableCode) {
		// 1.找到mapper中的所有path变量
        Map<String, String> paths = getPaths(config);
		if (log.isDebugEnabled()) {
			log.debug("execute:1.找到mapper中的所有path变量:{}", JSONUtil.toJsonStr(paths));
		}

        // 2.找出所有path的值，组成一个pathEntity对象，而data源数据中的最大数组长度决定了Entity的个数
        JSONArray result = JsonTreeProcessor.extractValuesToJsonArray(data, paths);

		if (log.isDebugEnabled()) {
			log.debug("execute:2.找出所有path的值:{}", JSONUtil.toJsonStr(result));
		}
        
        // 3.循环result，处理每一个pathEntity对象
		List<Entity> list = new ArrayList<>();
        for (int i = 0; i < result.size(); i++) {
			JSONObject pathEntity = result.getJSONObject(i);
			try {
				Entity entity = doLink(ds, dateTimeColumns, config, pathEntity, tableCode);
				if (entity != null) {
					list.add(entity);
				}
			} catch (Exception e) {
				log.error("映射源数据出现异常：" + e.getMessage(), e);
			}
		}
        
        return list;
	}
	
	/**
	 * 映射一个pathEntity对象到数据库
	 * @param ds 数据源
	 * @param dateTimeColumns 需要格式化的时间字段
	 * @param config 映射规则
	 * @param pathEntity 映射规则中所有path变量组成的一个Entity对象
	 * @param tableCode 表名
	 */
	private static Entity doLink(DataSource ds, List<String> dateTimeColumns, JSONObject config, 
			JSONObject pathEntity, String tableCode) {
		if (filter(config, pathEntity)) {
			// 映射出目标实体
			Entity entity = mapper(ds, dateTimeColumns, config, pathEntity, tableCode);
			
			// 执行每个字段的HlinkFunction
			JSONObject m = config.getJSONObject("mapper");
			for (String k : m.keySet()) {
				JSONObject field = m.getJSONObject(k);
				String function = field.getStr("func");
				Object oldValue = entity.get(k);
				entity.set(k, callFunction(ds, entity, function, oldValue));
			}
			
			return entity;
		}
		return null;
	}
	
	/**
	 * 根据路径源数据，映射出实体集合
	 * @param ds 数据源
	 * @param dateTimeColumns 需要格式化的时间字段
	 * @param config 映射规则
	 * @param pathEntity 源数据
	 * @param tableCode 表名
	 * @return
	 */
	private static Entity mapper(DataSource ds, List<String> dateTimeColumns, JSONObject config, 
			JSONObject pathEntity, String tableCode) {
		Entity entity = Entity.create(tableCode);
		JSONObject m = config.getJSONObject("mapper");
		for (String k : m.keySet()) {
			JSONObject field = m.getJSONObject(k);
			String mode = field.getStr("mode");
			String path = field.getStr("path");
			String value = field.getStr("value");
			JSONArray ruleArr = field.getJSONArray("rule");
			Object dataValue = null;
			
			if (StrUtil.isBlank(mode)) {
				continue;
			}
			
			switch (mode) {
			case "path":
				dataValue = pathEntity.get(path);
				break;
				
			case "value":
				dataValue = value;
				break;
				
			case "rule":
				dataValue = getValueByRule(pathEntity, ruleArr);
				break;

			default:
				break;
			}
			
			// 如果是时间字段，则格式化处理
			if (dateTimeColumns.contains(k) && dataValue != null) {
				entity.set(k, HtUtil.formatTimeString(dataValue.toString()));
			} else {
				entity.set(k, dataValue);
			}
		}
		
		if (log.isDebugEnabled()) {
			log.debug("mapper:完成映射。表：{}，规则：{}，源数据：{}，目标数据：{}", tableCode, JSONUtil.toJsonStr(config), 
					JSONUtil.toJsonStr(pathEntity), JSONUtil.toJsonStr(entity));
		}
		
		return entity;
	}

	/**
	 * 执行函数
	 * @param ds 数据源
	 * @param entity 映射后的新实体类
	 * @param function 函数实现类名称
	 * @param dataValue 函数参数
	 * @return
	 */
	private static Object callFunction(DataSource ds, Entity entity, String function, Object dataValue) {
		HlinkFunction<?> func = FunctionUtil.getFunction(function);
		if (func != null) {
			Object newValue = func.format(ds, dataValue, entity);
			log.debug("callFunction:执行函数：{}，原始值：{}，新值：{}", function, dataValue, newValue);
			return newValue;
		} else {
			log.debug("callFunction:函数为空：{}，不执行，直接返回原始值：{}", function, dataValue);
		}
		return dataValue;
	}

	/**
	 * 根据映射规则，从pathEntity中获取数据
	 * @param pathEntity 存储所有路径变量的值
	 * @param ruleArr 映射规则数组
	 * @return
	 */
	private static Object getValueByRule(JSONObject pathEntity, JSONArray ruleArr) {
		if (ruleArr != null && !ruleArr.isEmpty()) {
			for (int i = 0; i < ruleArr.size(); i++) {
				JSONObject rule = ruleArr.getJSONObject(i);
				FilterExpression filter = createFilterExpression(rule);
				if (JsonConditionEvaluator.evaluateCondition(filter, pathEntity)) {
					String thenValue = rule.getStr("then_value");
					String thenPath = rule.getStr("then_path");
					return thenValue != null ? thenValue : pathEntity.get(thenPath);
				}
			}
		}
		return null;
	}

	/**
	 * 获取config配置中所有带path的key
	 * @param mapper
	 * @return
	 */
	private static Map<String, String> getPaths(JSONObject mapper){
		Map<String, String> paths = new HashMap<>();
		
	    for (String key : mapper.keySet()) {
	        if (key.endsWith("path")) {
	        	String value = mapper.getStr(key);
	        	if (!paths.containsKey(value)) paths.put(value, value);
	        } else {
	            // 如果值是JSONObject，继续递归
	            Object value = mapper.get(key);
	            if (value instanceof JSONObject) {
	            	Map<String, String> childPaths = getPaths((JSONObject) value);
	            	childPaths.keySet().stream().forEach(k -> paths.put(k, k));
	            }
	            // 如果值是JSONArray，遍历数组中的JSONObject
	            else if (value instanceof JSONArray) {
	                JSONArray array = (JSONArray) value;
	                for (int i = 0; i < array.size(); i++) {
	                    if (array.get(i) instanceof JSONObject) {
	                    	Map<String, String> childPaths = getPaths((JSONObject) array.get(i));
	                    	childPaths.keySet().stream().forEach(k -> paths.put(k, k));
	                    }
	                }
	            }
	        }
	    }
		
		return paths;
	}

	/**
	 * 判断一个json源数据是否符合过滤条件
	 * @param config 过滤条件
	 * @param pathEntity 源数据
	 * @return
	 */
	private static boolean filter(JSONObject config, JSONObject pathEntity) {
        // 创建过滤条件
        FilterGroup filterGroup = new FilterGroup();
        List<FilterExpression> getConditions = getConditions(config, "filter");
        String filter_type = config.getStr("filter_type");
        if ("and".equals(filter_type)) {
        	filterGroup.setFilterAnd(getConditions);
        } else {
        	filterGroup.setFilterOr(getConditions);
        }
        
        // 评估条件
        return JsonConditionEvaluator.evaluate(filterGroup, pathEntity);
    }
	
	/**
	 * 从配置中获取过滤条件组
	 * @param mapper
	 * @param field
	 * @return
	 */
	private static List<FilterExpression> getConditions(JSONObject mapper, String field){
		List<FilterExpression> filters = new ArrayList<>();
		JSONArray arr = mapper.getJSONArray(field);
		if (arr != null) {
			for (int i = 0; i < arr.size(); i++) {
			    JSONObject c = arr.getJSONObject(i);
				filters.add(createFilterExpression(c));
			}
		}
		return filters;
	}

	/**
	 * 创建一个条件表达式类
	 */
	private static FilterExpression createFilterExpression(JSONObject c) {
		FilterExpression expr = new FilterExpression();
		expr.setLeftPath(c.getStr("left_path"));
		expr.setOperator(c.getStr("operator"));
		expr.setRightValue(c.getStr("right_value"));
		expr.setRightPath(c.getStr("right_path"));
		return expr;
	}
}
